package com.sequoaidb

import com.sequoiadb.base.{DBCollection, Sequoiadb}
import kafka.serializer.StringDecoder
import org.apache.log4j.Logger
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.bson.types.BasicBSONList
import org.bson.util.JSON
import org.bson.{BasicBSONObject, BSONObject}

import scala.util.Random


/**
 * Created by gaoshengjie on 2015/5/14.
 *
 *
 * brokers=192.168.30.221:9092
 * topics=mykafka
 * sdburl=
 * collectionspace=
 * collection=
 * seconds=
 */

object KafkaStreaming extends App{
  private[this] val logger = Logger.getLogger(getClass());
  /**
   * 把命令行参数解析到conf这个map中
   */
  var conf=Map[String,String]()
  //val Array(a,b,c)=args
  var t=  args.map(word=>{
    var param=word.split("=")
    conf += (param(0)->param(1))
  })
  Utils.validateArgs(conf)
  val brokers=conf.get("seconds").get
  val topics=conf.get("topics").get
  val sdburl=conf.get("sdburl").get
  val collectionspace=conf.get("collectionspace").get
  val collection=conf.get("collection").get
  val seconds=conf.get("seconds").get

  val topicsSet=topics.split(",").toSet
  val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
  val sparkConf=new SparkConf().setAppName("DirectKafkaWordCount")
  val ssc=new StreamingContext(sparkConf,Seconds(seconds.toInt))
  val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
    ssc, kafkaParams, topicsSet)
  val sdburlSet=sdburl.split(",")
  val lines = messages.map(_._2)
  lines.foreachRDD{
    rdd =>
      rdd.foreachPartition(
        partitionOfRecords => {
          val sequoiadb = new Sequoiadb(, "", "")
          val cspace = sequoiadb.getCollectionSpace(collectionspace)
          val cl = cspace.getCollection(collection)
          partitionOfRecords.foreach(
            record => {
              val bson=JSON.parse(record).asInstanceOf[BSONObject]
              val index= Random.nextInt(sdburlSet.size)
              val sequoiadb=new Sequoiadb(sdburlSet(index),"","")
              if(!sequoiadb.isCollectionSpaceExist(collectionspace))
              {
                sequoiadb.createCollectionSpace(collectionspace)
              }
              val cs=sequoiadb.getCollectionSpace(collectionspace)
              if(!cs.isCollectionExist(collection)){
                cs.createCollection(collection)
              }
              val cl=cs.getCollection(collection)
              process(bson,cl)
            }
          )
        }
      )
  }

  ssc.start()
  ssc.awaitTermination()


  def process(bson:BSONObject,cl:DBCollection): Unit ={


    val op=bson.get("OP")
    op match {
      case "I" => {
        var bsonObj=new BasicBSONObject()
        bsonObj.put("TNAME",bson.get("TNAME"));
        bsonObj.put("CSN",bson.get("CSN"));
        bsonObj.put("OSN",bson.get("OSN"));
        bsonObj.putAll(bsonObj.get("INSERT").asInstanceOf[BSONObject])
        cl.insert(bsonObj)
      }
      case "U" => {
        var andBson=new BasicBSONObject()
        var itemsArray=new BasicBSONList()
        itemsArray.add(new BasicBSONObject("TNAME",bson.get("TNAME")))
        itemsArray.add(new BasicBSONObject("CSN",new BasicBSONObject("$lte",bson.get("CSN"))))
        itemsArray.add(new BasicBSONObject("OSN",new BasicBSONObject("$lt",bson.get("OSN"))))

        var condtion=bson.get("COND").asInstanceOf[BSONObject].toMap()
        var it = condtion.entrySet().iterator()
        while(it.hasNext){
          var item=it.next()
          itemsArray.add(new BasicBSONObject(item.getKey().toString,item.getValue()))
        }

        var modifier=new BasicBSONObject()
        modifier.put("$set",bson.get("UPDATE").asInstanceOf[BSONObject])

        cl.update(andBson,modifier,new BasicBSONObject())
      }
      case "D" => {
        var andBson=new BasicBSONObject()
        var itemsArray=new BasicBSONList()
        itemsArray.add(new BasicBSONObject("TNAME",bson.get("TNAME")))
        itemsArray.add(new BasicBSONObject("CSN",new BasicBSONObject("$lte",bson.get("CSN"))))
        itemsArray.add(new BasicBSONObject("OSN",new BasicBSONObject("$lt",bson.get("OSN"))))
        var condtion=bson.get("COND").asInstanceOf[BSONObject].toMap
        var it = condtion.entrySet().iterator()
        while(it.hasNext){
          var item=it.next()
          itemsArray.add(new BasicBSONObject(item.getKey().toString,item.getValue()))
        }
        andBson.put("$and",itemsArray)
        cl.delete(andBson)
      }
      case _ => {
        logger.warn(bson.toString)
      }
    }
  }
}

